package com.test.reactor.control;

import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import com.test.reactor.config.GlobalException;

import akka.actor.ActorRef;
import akka.actor.ActorSelection;
import akka.actor.ActorSystem;
import reactor.core.publisher.Mono;
import scala.concurrent.duration.FiniteDuration;

@Controller
public class ActorController extends AbstractController{
	
	@Autowired
    private ActorSystem actorSystem;
	
	@Autowired
	private ActorRef receiveActorRef;
	
	@RequestMapping("actor/{who}")
	@ResponseBody
	public Mono<String> hello(@PathVariable String who) {
		
		ActorSelection actorSelection = actorSystem.actorSelection("akka://system/user/testActor");
		CompletionStage<ActorRef> resolveOneCS = actorSelection.resolveOneCS(new FiniteDuration(3, TimeUnit.MINUTES));
		Mono<String> flatMap = Mono.fromCompletionStage(resolveOneCS).doOnError(t -> {
			throw new GlobalException(HttpStatus.SERVICE_UNAVAILABLE, "my---"+t.getMessage());
		}).flatMap(act -> {
			System.out.println(act);
			act.tell("aaaaaaaaa", receiveActorRef);
			return success();
		});
		return flatMap;
	}
	
}
